package com.zzl.mq.network.gossip;

import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;


import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;


@Component
public class GossipManager {
    private static final Logger log = LoggerFactory.getLogger(GossipManager.class);
    
    private final Set<GossipMessageHandler> messageHandlers = new HashSet<>();
    private final ClusterNodeManager nodeManager;
    private final ScheduledExecutorService scheduler;
    private final Set<String> processedMessages;
    private final Map<String, GossipMessage> messageBuffer;
    
    @Value("${mq.network.gossip.fanout:3}")
    private int fanout;
    
    @Value("${mq.network.gossip.message-ttl:3600000}")
    private long messageTtl;
    
    @Value("${mq.network.gossip-interval:1000}")
    private long gossipInterval;

    @Value("${mq.network.node-id}")
    private String localNodeId;

    public GossipManager(ClusterNodeManager nodeManager) {
        this.nodeManager = nodeManager;
        this.scheduler = Executors.newScheduledThreadPool(1);
        this.processedMessages = Collections.newSetFromMap(new ConcurrentHashMap<>());
        this.messageBuffer = new ConcurrentHashMap<>();
    }
    
    @PostConstruct
    public void init() {
        startGossip();
        log.info("Gossip protocol initialized on application startup");
    }
    
    public void startGossip() {
        scheduler.scheduleAtFixedRate(this::gossipRound, 0, gossipInterval, TimeUnit.MILLISECONDS);
        scheduler.scheduleAtFixedRate(this::cleanupOldMessages, 0, gossipInterval * 10, TimeUnit.MILLISECONDS);
        log.info("Gossip protocol started with interval: {} ms", gossipInterval);
    }
    
    private void gossipRound() {
        List<ClusterNode> peers = nodeManager.getRandomPeers(fanout);
        if (peers.isEmpty()) {
            return;
        }
        
        // 发送消息缓冲区中的消息给随机选择的节点
        for (ClusterNode peer : peers) {
            Iterator<Map.Entry<String, GossipMessage>> iterator = messageBuffer.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, GossipMessage> entry = iterator.next();
                GossipMessage message = entry.getValue();
                try {
                    peer.sendGossipMessage(message);
                    log.debug("Sent gossip message to peer: {}", peer.getNodeId());
                    
                    // 确保消息被记录为已处理
                    processedMessages.add(message.getMessageId());
                    // 从缓冲区移除已发送的消息
                    iterator.remove();
                    
                } catch (Exception e) {
                    log.error("Failed to send gossip message to peer: " + peer.getNodeId(), e);
                }
            }
        }
    }
    
    public void spreadMessage(GossipMessage message) {
        if (processedMessages.contains(message.getMessageId())) {
            return;
        }
        
        message.setTimestamp(System.currentTimeMillis());
        messageBuffer.put(message.getMessageId(), message);
        processedMessages.add(message.getMessageId());
        
        log.debug("Spreading gossip message: {}", message.getMessageId());
    }
    
    public void handleGossipMessage(GossipMessage message) {
        // 如果消息已经处理过，直接返回
        if (processedMessages.contains(message.getMessageId())) {
            return;
        }

        log.info("Received expired gossip message: {}", System.currentTimeMillis()+messageTtl);
        // 检查消息是否过期
        if (System.currentTimeMillis() - message.getTimestamp() > messageTtl) {
            log.debug("Received expired gossip message: {}", message.getMessageId());
            return;
        }
        
        // 处理消息
        try {
            // 通知所有处理器
            for (GossipMessageHandler handler : messageHandlers) {
                handler.handleMessage(message);
            }
            
            // 将消息加入已处理集合和消息缓冲区，用于进一步传播
            processedMessages.add(message.getMessageId());
            messageBuffer.put(message.getMessageId(), message);
            
            log.debug("Handled gossip message: {}", message.getMessageId());
        } catch (Exception e) {
            log.error("Error handling gossip message: " + message.getMessageId(), e);
        }
    }
    
    private void cleanupOldMessages() {
        long now = System.currentTimeMillis();
        messageBuffer.entrySet().removeIf(entry -> 
            now - entry.getValue().getTimestamp() > messageTtl);
        log.debug("Cleaned up old messages from buffer");
    }
    
    @Scheduled(fixedDelayString = "${mq.network.gossip-interval}")
    public void sendHeartbeat() {
        GossipMessage heartbeatMessage = new GossipMessage();
        heartbeatMessage.setMessageId(UUID.randomUUID().toString());
        heartbeatMessage.setNodeId(localNodeId);
        heartbeatMessage.setType(GossipMessage.MessageType.HEARTBEAT);
        heartbeatMessage.setTimestamp(System.currentTimeMillis());
        
        //spreadMessage(heartbeatMessage);
    }
    
    public void addMessageHandler(GossipMessageHandler handler) {
        messageHandlers.add(handler);
    }
} 